package felix.fuse.mount.service;

import com.amazonaws.services.s3.model.ObjectListing;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Scheduler;
import jnr.ffi.Platform;
import jnr.ffi.Platform.OS;
import jnr.ffi.Pointer;
import jnr.ffi.types.mode_t;
import jnr.ffi.types.off_t;
import jnr.ffi.types.size_t;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.serce.jnrfuse.ErrorCodes;
import ru.serce.jnrfuse.FuseFillDir;
import ru.serce.jnrfuse.FuseStubFS;
import ru.serce.jnrfuse.struct.FileStat;
import ru.serce.jnrfuse.struct.FuseFileInfo;
import ru.serce.jnrfuse.struct.Statvfs;

import java.io.BufferedInputStream;
import java.io.File;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;

public class S3Fuse extends FuseStubFS implements BaseService {

    private final Logger log = LoggerFactory.getLogger(getClass());

    private final Cache<String, FsField> fsFieldCache = Caffeine.newBuilder().scheduler(Scheduler.forScheduledExecutorService(Executors.newScheduledThreadPool(2))).expireAfterAccess(Duration.ofMinutes(1)).maximumSize(1000 * 10).build();

    private final Cache<String, Object> listDirsCache = Caffeine.newBuilder().scheduler(Scheduler.forScheduledExecutorService(Executors.newScheduledThreadPool(2))).expireAfterWrite(Duration.ofSeconds(10))
            .maximumSize(1000 * 10).build();

    /**
     * s3桶，挂在目录起点是桶
     */
    private String rootBucket;

    /**
     * 修改文件或替换时，都是临时创建一个文件，写入后rename为目标文件
     * create /test/5/2/.goutputstream-TYH2L0 36
     * write /test/5/2/.goutputstream-TYH2L0 36 size:12 offset:0
     * rename oldPath:/test/5/2/.goutputstream-TYH2L0 newPath:/test/5/2/1.txt
     * flush /test/5/2/1.txt 36
     * release /test/5/2/1.txt 36
     * <p>
     * 处理逻辑是这样：
     * create命令时，把文件名和fid缓存到createFileCache
     * write写入的是.goutputstream-TYH2L0文件
     * rename时，flush .goutputstream-TYH2L0文件，完成该文件上传,然后覆盖掉原来的文件
     * flush 不做任何操作
     * release 不做任何操作
     * <p>
     * Duration.ofSeconds(50000)必须设置长一点，否则有可能上传超大文件时，由于时间过久，文件信息丢失。
     */
    private final Cache<String, Long> createFileCache = Caffeine.newBuilder().scheduler(Scheduler.forScheduledExecutorService(Executors.newScheduledThreadPool(2))).expireAfterWrite(Duration.ofSeconds(50000))
            .maximumSize(1000 * 10).build();

    private S3Service s3Service;

    private UploadFileToRemote uploadFileToRemote;

    // 10MB
    private static final long buffsetSize = 1024 * 1024 * 10;

    /**
     * 在挂在多个盘符的情况下，id不会重复
     */
    private static final AtomicLong id = new AtomicLong(0);

    private int file_mode = 0744;

    private int dir_mode = 0755;


    private boolean readOnly = false;


    public void setReadOnly(boolean readOnly) {
        this.readOnly = readOnly;
    }

    public void setRootBucket(String rootBucket) {
        this.rootBucket = rootBucket;
    }

    @Override
    public void init() {
        s3Service = ContextContainer.get(S3Service.class);
        uploadFileToRemote = ContextContainer.get(UploadFileToRemote.class);
        if (Platform.getNativePlatform().getOS() == OS.WINDOWS) {
            file_mode = 0777;
            dir_mode = 0777;
        }
    }

    @Override
    public void destroy() {
        s3Service.close();
    }


    public boolean connection(String accessKey, String secretKey, String endpoint) {
        s3Service.set(accessKey, secretKey, endpoint);
        return this.reConnectS3();
    }

    public boolean isInitS3Service() {
        return s3Service.isInit();
    }

    public boolean reConnectS3() {
        return s3Service.reConnectS3();
    }

    @Override
    public int getattr(String path, FileStat stat) {
        if (path.startsWith("/.svn") || path.endsWith("/.svn") || path.contains("/.svn/") || path.contains("/desktop.ini")) {
            return -ErrorCodes.ENOENT();
        }
        log.trace("getattr {}", path);
        FsField fsField = this.getFsField(path);
        if (fsField == null) {
            return -ErrorCodes.ENOENT();
        }

        stat.st_nlink.set(fsField.st_nlink);
        stat.st_uid.set(getContext().uid.get());
        stat.st_gid.set(getContext().gid.get());
        if (fsField.isFile) {
            stat.st_mode.set(fsField.st_mode | this.file_mode);
            stat.st_size.set(fsField.st_size);
            long ctime_sec = fsField.st_mtim / 1000;
            long ctime_nsec = (fsField.st_mtim % 1000) * 1000;
            stat.st_mtim.tv_sec.set(ctime_sec);
            stat.st_mtim.tv_nsec.set(ctime_nsec);
        } else {
            stat.st_mode.set(fsField.st_mode | this.dir_mode);
        }
        return 0;
    }

    @Override
    public int statfs(String path, Statvfs stbuf) {
        if (Platform.getNativePlatform().getOS() == OS.WINDOWS) {
            if ("/".equals(path)) {
                stbuf.f_bsize.set(1024000);
                stbuf.f_frsize.set(102400);
                stbuf.f_blocks.set(1024 * 10240000);
                stbuf.f_bfree.set(1024 * 102400);
            }
        }
        return super.statfs(path, stbuf);
    }

    @Override
    public int readdir(String path, Pointer buf, FuseFillDir filter, @off_t long offset, FuseFileInfo fi) {
        log.debug("readdir {} {} offset:{}", path, fi.fh.get(), offset);
        filter.apply(buf, ".", null, 0);
        filter.apply(buf, "..", null, 0);
        ObjectListing objectListing = (ObjectListing) listDirsCache.getIfPresent(path);
        if (objectListing == null) {
            objectListing = s3Service.listSub(rootBucket, path);
            if (objectListing == null) {
                return 0;
            }
            listDirsCache.put(path, objectListing);
        }
        List<S3ObjectSummary> s3ObjectSummaryList = objectListing.getObjectSummaries();
        List<String> commonPrefixes = objectListing.getCommonPrefixes();
        for (S3ObjectSummary s3ObjectSummary : s3ObjectSummaryList) {
            String key = s3ObjectSummary.getKey();
            String name = this.getFileName(key);
            if (name.equals(S3Service.empty)) {
                continue;
            }
            FsField fsField = this.s3ObjectSummaryToFsField(s3ObjectSummary);
            String keyPath = null;
            if (path.endsWith("/")) {
                keyPath = path + name;
            } else {
                keyPath = path + "/" + name;
            }
            fsFieldCache.put(keyPath, fsField);
            filter.apply(buf, name, null, 0);
        }
        for (String commonPrefix : commonPrefixes) {
            String name = this.getFileName(commonPrefix);
            FsField fsField = this.getDirectoryFsField();
            String keyPath = null;
            if (path.endsWith("/")) {
                keyPath = path + name;
            } else {
                keyPath = path + "/" + name;
            }
            fsFieldCache.put(keyPath, fsField);
            filter.apply(buf, name, null, 0);
        }
        return 0;
    }

    @Override
    public int open(String path, FuseFileInfo fi) {
        fi.fh.set(id.getAndIncrement());
        log.debug("open {}", fi.fh.get());
        return 0;
    }

    /**
     * 在linux下，这个会存在多个并发读，所以无法使用缓存之类的办法，暂时不做任何缓存处理
     *
     * @param path
     * @param buf
     * @param size
     * @param offset
     * @param fi
     * @return
     */
    @Override
    public int read(String path, Pointer buf, @size_t long size, @off_t long offset, FuseFileInfo fi) {
        try {
            FsField fsField = this.getFsField(path);
            if (fsField == null) {
                return -ErrorCodes.ENOENT();
            }
            if (offset >= fsField.st_size) {
                return 0;
            }
            log.debug("read: {} {} {}", path, size, offset);
            byte[] remoteBytes = s3Service.read(rootBucket, path, size, offset);
            buf.put(0, remoteBytes, 0, remoteBytes.length);
            return remoteBytes.length;
//            if (!this.checkPeriod(size, offset, fi)) {
//                log.info("readBuffer: {} {} {} {}",fi.fh.get(),offset,size,this);
//                byte[] bytes = this.readBuffer(fsField, path, size, offset, fi);
//                buf.put(0, bytes, 0, bytes.length);
//                return bytes.length;
//            } else {
//                int rd = 0;
//                int nread = 0;
//                int sz = (int) size;
//                log.info("streaming: {} {} {} {}",fi.fh.get(),offset,size,this);
//                BufferedInputStream in = this.openFile(path, fi, offset, size);
//                final byte[] dest = new byte[sz];
//                while (rd >= 0 && nread < size) {
//                    rd = in.read(dest, nread, sz - nread);
//                    if (rd >= 0) {
//                        nread += rd;
//                    }
//                }
//
//                if (nread == -1) { // EOF
//                    nread = 0;
//                } else if (nread > 0) {
//                    buf.put(0, dest, 0, nread);
//                }
//                return nread;
//            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return -ErrorCodes.ENOENT();
        }
    }

    @Override
    public int create(String path, @mode_t long mode, FuseFileInfo fi) {
        if (this.readOnly) {
            return -ErrorCodes.EACCES();
        }
        String pPath = this.getParentPath(path);
        if (pPath == null) {
            return -ErrorCodes.EACCES();
        }
        Long fid = id.getAndIncrement();
        fi.fh.set(fid);
        log.debug("create {} {} mode:{}", path, fi.fh.get(), mode);
        if (s3Service.fileExist(rootBucket, path)) {
            return -ErrorCodes.EEXIST();
        }
        createFileCache.put(path, fid);
        FsField fsField = new FsField();
        fsField.st_mode = FileStat.S_IFREG;
        fsField.st_nlink = 1;
        fsField.st_size = 0L;
        fsField.isFile = true;
        fsField.st_mtim = new Date().getTime();
        fsFieldCache.put(path, fsField);
        return 0;
    }

    @Override
    public int write(String path, Pointer buf, @size_t long size, @off_t long offset,
                     FuseFileInfo fi) {
        log.debug("write {} {} size:{} offset:{}", path, fi.fh.get(), size, offset);
        final int sz = (int) size;
        final long fd = fi.fh.get();
        byte[] buffer = new byte[sz];
        try {
            buf.get(0, buffer, 0, sz);
            boolean isWrite = uploadFileToRemote.write(rootBucket, path, fd, buffer);
            if (isWrite) {
                return sz;
            } else {
                return -ErrorCodes.EIO();
            }
        } catch (Exception e) {
            log.error("while writing to {}.", path, e);
            return -ErrorCodes.EIO();
        }

    }

    @Override
    public int flush(String path, FuseFileInfo fi) {
        log.debug("flush {} {}", path, fi.fh.get());
        final long fd = fi.fh.get();
        uploadFileToRemote.flush(rootBucket, path, fd);
        String pPath = this.getParentPath(path);
        listDirsCache.invalidate(pPath);
        fsFieldCache.invalidate(path);
        return 0;
    }

    @Override
    public int mkdir(String path, @mode_t long mode) {
        if (this.readOnly) {
            return -ErrorCodes.EACCES();
        }
        String pPath = this.getParentPath(path);
        if (pPath == null) {
            return -ErrorCodes.EACCES();
        }
        log.debug("mkdir {} mode:{}", path, mode);
        s3Service.mkdir(rootBucket, path);
        return 0;
    }

    @Override
    public int rename(String oldPath, String newPath) {
        if (this.readOnly) {
            return -ErrorCodes.EACCES();
        }
        String pNewPath = this.getParentPath(newPath);
        if (pNewPath == null) {
            return -ErrorCodes.EACCES();
        }
        log.debug("rename oldPath:{} newPath:{}", oldPath, newPath);
        String pOldPath = this.getParentPath(oldPath);
        Long fid = this.createFileCache.getIfPresent(oldPath);
        if (fid != null) {
            uploadFileToRemote.flush(rootBucket, oldPath, fid);
            this.createFileCache.invalidate(oldPath);
        }
        FsField fsField = this.getFsField(oldPath);
        if (fsField == null) {
            return -ErrorCodes.ENOTDIR();
        }
        if (fsField.isFile) {
            s3Service.renameFile(rootBucket, oldPath, newPath);
        } else {
            s3Service.renameDir(rootBucket, oldPath, newPath);
        }
        listDirsCache.invalidate(pNewPath);
        listDirsCache.invalidate(pOldPath);
        fsFieldCache.invalidate(oldPath);
        return 0;
    }

    @Override
    public int rmdir(String path) {
        if (this.readOnly) {
            return -ErrorCodes.EACCES();
        }
        log.debug("rmdir {}", path);
        String pPath = this.getParentPath(path);
        if (pPath == null) {
            return -ErrorCodes.EACCES();
        }
        boolean isSuc = s3Service.rmdir(rootBucket, path);
        if (isSuc) {
            listDirsCache.invalidate(pPath);
            fsFieldCache.invalidate(path);
            return 0;
        } else {
            return ErrorCodes.EIO();
        }
    }

    @Override
    public int unlink(String path) {
        if (this.readOnly) {
            return -ErrorCodes.EACCES();
        }
        log.debug("unlink {}", path);
        boolean isSuc = s3Service.deleteObject(rootBucket, path);
        if (isSuc) {
            String pPath = this.getParentPath(path);
            listDirsCache.invalidate(pPath);
            fsFieldCache.invalidate(path);
            return 0;
        } else {
            return ErrorCodes.EIO();
        }
    }


    @Override
    public int release(String path, FuseFileInfo fi) {
        log.debug("release {} {}", path, fi.fh.get());
        Long id = fi.fh.get();
        this.createFileCache.invalidate(path);
        this.fsFieldCache.invalidate(id);
        return 0;
    }


    private String getFileName(String key) {
        return new File(key).getName();
    }

    private boolean isRoot(String path) {
        return new File(path).getParent() == null;
    }

    private String getParentPath(String path) {
        String parent = new File(path).getParent();
        if (parent != null) {
            return parent.replaceAll("\\\\", "/");
        } else {
            return null;
        }
    }

    private FsField getFsField(String path) {
        FsField fsField = fsFieldCache.getIfPresent(path);
        if (fsField != null) {
            return fsField;
        }
        String parent = this.getParentPath(path);
        // 请求的path是根/
        if (parent == null) {
            fsField = this.getDirectoryFsField();
            return fsField;
        }
        ObjectListing objectListing = s3Service.listSub(rootBucket, parent);
        List<S3ObjectSummary> s3ObjectSummaryList = objectListing.getObjectSummaries();
        List<String> commonPrefixes = objectListing.getCommonPrefixes();
        for (S3ObjectSummary s3ObjectSummary : s3ObjectSummaryList) {
            String key = s3ObjectSummary.getKey();
            String name = this.getFileName(key);
            fsField = this.s3ObjectSummaryToFsField(s3ObjectSummary);
            if (parent.endsWith("/")) {
                fsFieldCache.put(parent + name, fsField);
            } else {
                fsFieldCache.put(parent + "/" + name, fsField);
            }
        }
        for (String commonPrefix : commonPrefixes) {
            String name = this.getFileName(commonPrefix);
            fsField = this.getDirectoryFsField();
            if (parent.endsWith("/")) {
                fsFieldCache.put(parent + name, fsField);
            } else {
                fsFieldCache.put(parent + "/" + name, fsField);
            }
        }
        fsField = fsFieldCache.getIfPresent(path);
        return fsField;
    }

    private FsField s3ObjectSummaryToFsField(S3ObjectSummary s3ObjectSummary) {
        FsField fsField = new FsField();
        fsField.st_mode = FileStat.S_IFREG;
        fsField.st_nlink = 1;
        fsField.isFile = true;
        fsField.st_size = s3ObjectSummary.getSize();
        fsField.st_mtim = s3ObjectSummary.getLastModified().getTime();
        return fsField;
    }

    private FsField getDirectoryFsField() {
        FsField fsField = new FsField();
        fsField.st_mode = FileStat.S_IFDIR;
        fsField.isFile = false;
        fsField.st_nlink = 1;
        return fsField;
    }

    public class FsField {
        public int st_mode;
        public int st_nlink;
        public Long st_mtim = -1L;
        public Long st_size = -1L;
        public boolean isFile;
    }

    public class RemoteStream {
        public BufferedInputStream in;
        public S3Object s3Ojbect;
    }

    public class FileOffset {
        public Long offset;
        public byte[] bytes;
    }

    public class PeriodOffsetCache {
        public Long lastOffset;
        public int cnt;
    }

    public void mount(String path) {
        this.mount(Paths.get(path), false, false, new String[]{"-osync_read", "-odirect_io", "-s"});
    }

    @Override
    public void umount() {
        super.umount();
        s3Service.close();
    }
}
